/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.nhncorp.neptune.changelog;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.nhncorp.neptune.common.Constants;
import com.nhncorp.neptune.common.LockServiceFactory;
import com.nhncorp.neptune.common.conf.NConfiguration;
import com.nhncorp.neptune.common.exception.ChangeLogException;
import com.nhncorp.neptune.common.util.NetworkUtil;
import com.nhncorp.neptune.fs.GPath;
import com.nhncorp.neptune.fs.NeptuneFileSystem;
import com.nhncorp.pleiades.client.LockEventData;
import com.nhncorp.pleiades.client.LockEventFilter;
import com.nhncorp.pleiades.client.LockEventHandler;
import com.nhncorp.pleiades.client.LockEventMonitor;
import com.nhncorp.pleiades.client.LockService;
import com.nhncorp.pleiades.protocol.LockMetaData;

public class ServerLocationManager {
  static final Log LOG = LogFactory.getLog(ServerLocationManager.class);
  private static Object singletonLock = new Object();
  private static ServerLocationManager singleton = null;

  public static ServerLocationManager instance(NConfiguration conf) throws IOException {
    synchronized (singletonLock) {
      if (singleton == null) {
        if (conf == null) {
          throw new IOException("NConfiguration instance is required " 
              + "to initiate singleton instance of ServerLocationManager");
        }
        singleton = new ServerLocationManager(conf);
      }

      singleton.loadFromPleiades();
      return singleton;
    }
  }

  private String logImagePath;
  private int numReplicas;

  ReentrantReadWriteLock listManagerLock = new ReentrantReadWriteLock();
  ServerListManagementStrategy listManager = new RandomStrategy();

  Map<String, InetSocketAddress[]> cache = new HashMap<String, InetSocketAddress[]>();

  InetSocketAddress localChangeLogServerAddress;
  AtomicBoolean hasAlreadyBeenLoaded = new AtomicBoolean(false);

  private NeptuneFileSystem fs;
  LockService lockService;

  private ServerLocationManager(NConfiguration conf) throws IOException {
    String owner = "LocationManager:" 
      + InetAddress.getLocalHost().getHostName() + ":" 
      + conf.getInt("tabletServer.port", 7001);
    try {
      lockService = LockServiceFactory.getLockService(conf, owner, true);
    } catch(IOException e) {
      LOG.fatal("Cannot get lock service with owner[" + owner + "]");
      throw e;
    }

    this.logImagePath = conf.get("changelog.image.dir", "/user/neptune/changelog");
    this.numReplicas = conf.getInt("changelog.num.replicas", 3);

    loadFromPleiades();
    initNeptuneFileSystem(conf);
  }

  public void clear() {
    synchronized (cache) {
      cache.clear();
    }
    hasAlreadyBeenLoaded.set(false);
    listManager.clear();
  }

  public InetSocketAddress[] getAllAddresses() {
    listManagerLock.readLock().lock();
    try {
      return listManager.getAllLogServerAddressList();
    } finally {
      listManagerLock.readLock().unlock();
    }
  }

  public InetSocketAddress[] getReplicaAddresses(String tabletName) throws IOException {
    InetSocketAddress[] addrList = null;

    try {
      synchronized (cache) {
        if ((addrList = cache.get(tabletName)) == null) {
          loadCacheFromImage(tabletName);

          if ((addrList = cache.get(tabletName)) == null) {
            listManagerLock.readLock().lock();
            try {
              addrList = listManager.electChangeLogServers();
            } finally {
              listManagerLock.readLock().unlock();
            }
            cache.put(tabletName, addrList);
            storeToImage(tabletName, addrList);
          }
        }
      }
    } catch (IOException e) {
      LOG.error(e);
      throw e;
    }

    return addrList;
  }

  public void removeReplica(String tabletName) throws IOException {
    LOG.info("remove change log server info of tablet[" + tabletName + "]");

    removeFromImage(tabletName);

    synchronized (cache) {
      InetSocketAddress[] removedAddresses = null; 
      
      if ((removedAddresses = cache.remove(tabletName)) != null 
          && localChangeLogServerAddress != null) {
        for(InetSocketAddress addr : removedAddresses) {
          if (addr.equals(localChangeLogServerAddress)) {
            localChangeLogServerAddress = null;
          }
        }
      }
    }
  }

  private void removeFromImage(String tabletName) throws IOException {
    GPath changeLogMetaPath = new GPath(logImagePath + "/" + tabletName);

    if (fs.exists(changeLogMetaPath)) {
      fs.delete(changeLogMetaPath);
    }
  }

  private void storeToImage(String tabletName, InetSocketAddress[] addrList) throws IOException {
    GPath changeLogMetaPath = new GPath(logImagePath + "/" + tabletName);

    if (fs.exists(changeLogMetaPath)) {
      if (isSameChangeLogInfo(changeLogMetaPath, addrList)) {
        LOG.info("Ignore deleting change log meta path: " + (logImagePath + "/" + tabletName));
        return;
      } else {
        LOG.info("Deleting change log meta path: " + (logImagePath + "/" + tabletName)
            + " and remake");
        fs.delete(changeLogMetaPath, true);
      }
    }

    // META 정보 생성
    OutputStream out = fs.create(changeLogMetaPath);
    for (InetSocketAddress eachServer : addrList) {
      out.write((eachServer.getHostName() + ":" + eachServer.getPort() + "\n").getBytes());
    }
    out.close();
  }

  private boolean isSameChangeLogInfo(GPath changeLogMetaPath, InetSocketAddress[] addrList)
      throws IOException {
    BufferedReader br = null;

    try {
      br = new BufferedReader(new InputStreamReader(fs.open(changeLogMetaPath)));
      String line = null;

      while ((line = br.readLine()) != null) {
        int i = 0;
        for (; i < addrList.length; i++) {
          if (addrList.equals(NetworkUtil.getAddress(line))) {
            break;
          }
        }

        if (i == addrList.length) {
          return false;
        }
      }
    } finally {
      if (br != null) {
        br.close();
      }
    }

    return true;
  }

  private void loadCacheFromImage(String tabletName) throws IOException {
    GPath tabletPath = new GPath(logImagePath, tabletName);
    if (fs.exists(tabletPath)) {
      BufferedReader br = null;
      try {
        br = new BufferedReader(new InputStreamReader(fs.open(tabletPath)));
        String line = null;
        List<InetSocketAddress> addrList = new ArrayList<InetSocketAddress>();
        while ((line = br.readLine()) != null) {
          addrList.add(NetworkUtil.getAddress(line));
        }

        synchronized (cache) {
          cache.put(tabletName, addrList.toArray(new InetSocketAddress[0]));
        }
      } finally {
        if (br != null) {
          br.close();
        }
      }
    }
  }

  private void initNeptuneFileSystem(NConfiguration conf) throws IOException {
    fs = NeptuneFileSystem.get(conf);
    GPath parentPath = new GPath(logImagePath);

    if (!fs.exists(parentPath)) {
      fs.mkdirs(parentPath);
    }
  }

  private void loadFromPleiades() throws IOException {
    if (hasAlreadyBeenLoaded.get()) {
      return;
    }

    listManagerLock.writeLock().lock();

    if (hasAlreadyBeenLoaded.get()) {
      return;
    }

    try {
      if (listManager.isEmpty() == false) {
        listManager.clear();
      }
      
      Map<String, LockMetaData> dirs = lockService.readDir(Constants.CHANGELOG_SERVER);
      if (dirs == null) {
        throw new ChangeLogException("No live change log servers");
      } else if (dirs.size() < numReplicas) {
        throw new ChangeLogException("Insufficient number of change log servers. "
            + "According to neptune configuration, changelog.num.replicas is " + numReplicas);
      }

      for (LockMetaData eachMetaData : dirs.values()) {
        if (!lockService.tryLock(eachMetaData.getLockId(), LockService.LockMode.LOCK_EXCLUSIVE)) {
          try {
            listManager.addChangeLogServers(NetworkUtil.fileNameToHostName(LockServiceFactory
                .getLockMetaDataPathName(eachMetaData)));
          } catch (IOException e) {
            LOG.warn("Fail to add change log server ["
                + NetworkUtil.fileNameToHostName(LockServiceFactory
                    .getLockMetaDataPathName(eachMetaData)) + "]");
            continue;
          } 
        } else {
          LOG.warn("check ChangeLogServer [" + NetworkUtil.fileNameToHostName(LockServiceFactory
              .getLockMetaDataPathName(eachMetaData)) + "]. it looks abnormal");
        }
      }

      hasAlreadyBeenLoaded.set(true);

      if (lockService.getEventHandler() != null) {
        LockEventHandler handler = lockService.getEventHandler();
        handler.clearLockEventMonitors();
        handler.addLockEventMonitor(createNodeRemovedEventMonitor(Constants.CHANGELOG_SERVER));
        handler.addLockEventMonitor(createNodeAddedEventMonitor(Constants.CHANGELOG_SERVER));
        lockService.setEventHandler(handler);
      }
    } finally {
      listManagerLock.writeLock().unlock();
    }
  }

  private LockEventMonitor createNodeRemovedEventMonitor(final String lockPath) {
    return new LockEventMonitor() {
      public LockEventFilter getFilter() {
        LockEventFilter filter = new LockEventFilter();
        filter.addConstraint(LockEventFilter.Operator.PF, lockPath);

        List<LockService.Events> lockEvents = new ArrayList<LockService.Events>();
        lockEvents.add(LockService.Events.NODE_REMOVED);
        filter.setLockEventList(lockEvents);

        return filter;
      }

      public void notifyEvent(LockService.Events eventType, LockEventData eventData) {
        if (!eventData.getLockId().equals(lockPath)) {
          String logServer = NetworkUtil.fileNameToHostName(LockServiceFactory
              .getLockPathName(eventData));

          LOG.info("change log server node [" + logServer + "] is deleted");
          try {
            listManager.removeChangeLogServer(logServer);
          } catch (IOException e) {
            LOG.warn("removing change log server [" + logServer + "] from list manager", e);
          }
          shrinkAddressCache(logServer);
        }
      }

      private void shrinkAddressCache(String logServer) {
        synchronized (cache) {
          String tabletName = findTabletName(logServer);

          if (tabletName == null) {
            return;
          }

          InetSocketAddress[] addrList = cache.get(tabletName);
          InetSocketAddress[] newAddrList = new InetSocketAddress[addrList.length - 1];

          for (int index = 0, i = 0; i < addrList.length; i++) {
            if (addrList[i].getHostName().equals(logServer) == false) {
              newAddrList[index++] = addrList[i];
            }
          }

          cache.put(tabletName, newAddrList);
        }
      }

      private String findTabletName(String logServer) {
        for (Map.Entry<String, InetSocketAddress[]> entry : cache.entrySet()) {
          for (int i = 0; i < entry.getValue().length; i++) {
            if (entry.getValue()[i].getHostName().equals(logServer)) {
              return entry.getKey();
            }
          }
        }

        return null;
      }
    };
  }

  private LockEventMonitor createNodeAddedEventMonitor(final String lockPath) {
    return new LockEventMonitor() {
      public LockEventFilter getFilter() {
        LockEventFilter filter = new LockEventFilter();
        filter.addConstraint(LockEventFilter.Operator.PF, lockPath);

        List<LockService.Events> lockEvents = new ArrayList<LockService.Events>();
        lockEvents.add(LockService.Events.NODE_ADDED);
        filter.setLockEventList(lockEvents);

        return filter;
      }

      public void notifyEvent(LockService.Events eventType, LockEventData eventData) {
        String logServer = NetworkUtil.fileNameToHostName(LockServiceFactory
            .getLockPathName(eventData));

        LOG.info("change log server node [" + logServer + "] is added");

        try {
          listManager.addChangeLogServers(logServer);
        } catch (IOException e) {
          LOG.warn("adding change log server [" + logServer + "] from list manager", e);
        }
      }
    };
  }

  // // Server List Management Strategies ////

  abstract class ServerListManagementStrategy {
    protected ArrayList<InetSocketAddress> logServerAddrList = new ArrayList<InetSocketAddress>();

    abstract InetSocketAddress[] electChangeLogServers() throws IOException;

    abstract void removeChangeLogServer(String hostAddr) throws IOException;

    abstract void addChangeLogServers(String hostAddr) throws IOException;

    public InetSocketAddress[] getAllLogServerAddressList() {
      return logServerAddrList.toArray(new InetSocketAddress[logServerAddrList.size()]);
    }

    protected int getPort(String addr) {
      return Integer.valueOf(addr.substring(addr.indexOf(":") + 1));
    }

    protected String getHost(String addr) {
      return addr.substring(0, addr.indexOf(":"));
    }

    public boolean isEmpty() {
      return logServerAddrList.isEmpty();
    }

    public void clear() {
      logServerAddrList.clear();
    }
  }

  class RandomStrategy extends ServerListManagementStrategy {
    public void addChangeLogServers(String hostAddrStr) throws IOException {
      InetSocketAddress hostAddr = new InetSocketAddress(InetAddress
          .getByName(getHost(hostAddrStr)), getPort(hostAddrStr));

      if (hostAddr.getAddress().equals(InetAddress.getLocalHost())) {
        localChangeLogServerAddress = hostAddr;
      }

      LOG.info("ChangeLogServer [" + hostAddrStr + "] is ready");
      synchronized (logServerAddrList) {
        logServerAddrList.add(hostAddr);
      }
    }

    public InetSocketAddress[] electChangeLogServers() throws IOException {
      InetSocketAddress[] ret = new InetSocketAddress[numReplicas];

      if (numReplicas > logServerAddrList.size()) {
        throw new IOException("Cannot elect changelog server!! the num of available servers are "
            + logServerAddrList.size());
      }

      int i = 0;
      if (localChangeLogServerAddress != null) {
        ret[0] = localChangeLogServerAddress;
        i = 1;
      }

      int prevIndex = -1;
      while (i < numReplicas) {
        int index = Math.abs(new Random().nextInt()) % logServerAddrList.size();

        if (prevIndex == index) {
          continue;
        }

        InetSocketAddress selected = logServerAddrList.get(index);
        prevIndex = index;
        if (isNew(ret, selected)) {
          ret[i++] = selected;
        }
      }

      return ret;
    }

    private boolean isNew(InetSocketAddress[] addresses, InetSocketAddress selected) {
      for (int i = 0; i < addresses.length; i++) {
        if (addresses[i] == null) {
          return true;
        } else if (addresses[i].equals(selected)) {
          return false;
        }
      }
      return true;
    }

    public void removeChangeLogServer(String hostAddrStr) throws IOException {
      InetSocketAddress hostAddr = new InetSocketAddress(InetAddress
          .getByName(getHost(hostAddrStr)), getPort(hostAddrStr));

      synchronized (logServerAddrList) {
        logServerAddrList.remove(hostAddr);
      }
    }
  }
}
